//package cn.iocoder.yudao.module.liveprice.kafka.consumer;
//
//import cn.iocoder.yudao.framework.common.enums.metal.MetalTypeEnum;
//import cn.iocoder.yudao.module.infra.api.websocket.WebSocketSenderApi;
//import cn.iocoder.yudao.module.liveprice.api.LivePriceApi;
//import cn.iocoder.yudao.module.liveprice.api.dto.LivePriceRespDTO;
//import cn.iocoder.yudao.module.liveprice.apihttp.NowApiTypeEnum;
//import cn.iocoder.yudao.module.liveprice.config.KafkaTopicProperties;
//import cn.iocoder.yudao.module.liveprice.kafka.NowApiPriceMessage;
//import com.alibaba.fastjson.JSONObject;
//import jakarta.annotation.Resource;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.context.annotation.Bean;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.listener.KafkaListenerErrorHandler;
//import org.springframework.stereotype.Component;
//
//import java.math.BigDecimal;
//import java.util.List;
//import java.util.Objects;
//import java.util.stream.Collectors;
//
//@Component
//@Slf4j
//public class NowApiPriceConsumer {
//
//    private static final String LIVE_MESSAGE_PRICE = "live_price";
//
//    @Resource
//    private KafkaTopicProperties kafkaTopicProperties;
//
//    @Resource
//    private LivePriceApi livePriceApi;
//    @Resource
//    private WebSocketSenderApi webSocketSenderApi;
//
//    @KafkaListener(
//            topics = "#{@nowApiPriceConsumer.getTopics()}",
//            groupId = "#{@nowApiPriceConsumer.getGroupId()}",
//            errorHandler = "nowApiKafkaListenerErrorHandler",
//            containerFactory = "nowApiKafkaListenerContainerFactory")
//    public void onMessage(NowApiPriceMessage message) {
//        handleData(message);
//    }
//
//    @Bean("nowApiKafkaListenerErrorHandler")
//    public KafkaListenerErrorHandler nowApiKafkaListenerErrorHandler() {
//        return (message, exception) -> {
//            log.error("NowApi Error processing record: " + message, exception);
//            return message;
//        };
//    }
//
//    public String[] getTopics() {
//        return kafkaTopicProperties.getNowapi().getCodes().stream()
//                .map(code -> kafkaTopicProperties.getNowapi().getPrefix() + code)
//                .toArray(String[]::new);
//    }
//
//    public String getGroupId() {
//        return kafkaTopicProperties.getNowapi().getGroupId();
//    }
//
//    /**
//     * 处理NowApi接口价格数据，进行计算返回
//     * @param message
//     * @return
//     */
//    private void handleData(NowApiPriceMessage message) {
//        NowApiTypeEnum typeEnum = Objects.requireNonNull(NowApiTypeEnum.getType(message.getGoldid()));
//
//        LivePriceRespDTO data = getConfPrice(message, typeEnum.getMaterialType());
//
//        // 异步写入Redis缓存
//        livePriceApi.asyncCachePriceData(data);
//
//        webSocketSenderApi.send(LIVE_MESSAGE_PRICE, JSONObject.toJSONString(data), typeEnum.getMaterialType());
//    }
//
//    /**
//     * 处理价格数据
//     * @param message NowApi价格消息
//     * @param metalType 金属类型
//     * @return 处理后的价格数据
//     */
//    private LivePriceRespDTO getConfPrice(NowApiPriceMessage message, Integer metalType) {
//        MetalTypeEnum metal = MetalTypeEnum.valueOf(metalType);
//        LivePriceRespDTO priceRespDTO = new LivePriceRespDTO();
//        priceRespDTO.setMetalName(metal.getTag());
//        priceRespDTO.setMetalType(metal.getType());
//
//        // 实际价格
//        priceRespDTO.setReclaimPrice(new BigDecimal(message.getBuy_price()));
//        priceRespDTO.setSalePrice(new BigDecimal(message.getSell_price()));
//        priceRespDTO.setHighPrice(new BigDecimal(message.getSell_price()));
//        priceRespDTO.setLowPrice(new BigDecimal(message.getLow_price()));
//        priceRespDTO.setOpenPrice(new BigDecimal(message.getOpen_price()));
//        priceRespDTO.setYesyPrice(new BigDecimal(message.getYesy_price()));
//
//        return priceRespDTO;
//    }
//
//}